package com.ruyuan.rapid.core.netty.processor;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.ruyuan.rapid.common.concurrent.queue.mpmc.MpmcBlockingQueue;
import com.ruyuan.rapid.common.enums.ResponseCode;
import com.ruyuan.rapid.core.RapidConfig;
import com.ruyuan.rapid.core.context.HttpRequestWrapper;
import com.ruyuan.rapid.core.helper.ResponseHelper;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.http.*;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Mpmc的核心实现处理器
 * @author mac
 */
@Slf4j
public class NettyMpmcProcessor implements NettyProcessor {

    private RapidConfig rapidConfig;

    private NettyCoreProcessor nettyCoreProcessor;

    private MpmcBlockingQueue<HttpRequestWrapper> mpmcBlockingQueue;

    private boolean usedExecutorPool;

    /**
     * 线程池
     */
    private ExecutorService executorService;

    private volatile boolean isRunning = false;

    private Thread consumerProcessorThread;

    public NettyMpmcProcessor(RapidConfig rapidConfig, NettyCoreProcessor nettyCoreProcessor, boolean usedExecutorPool) {
        this.rapidConfig = rapidConfig;
        this.nettyCoreProcessor = nettyCoreProcessor;
        this.mpmcBlockingQueue = new MpmcBlockingQueue<>(rapidConfig.getBufferSize());
        this.usedExecutorPool = usedExecutorPool;
    }

    /**
     * 核心执行方法
     * @param httpRequestWrapper
     */
    @Override
    public void process(HttpRequestWrapper httpRequestWrapper) throws Exception {
        System.out.println("NettyMpmcProcessor put ok......");
        this.mpmcBlockingQueue.put(httpRequestWrapper);
    }

    /**
     * 执行器启动方法
     */
    @Override
    public void start() {
        nettyCoreProcessor.start();
        this.isRunning = true;
        if (usedExecutorPool) {
            executorService = Executors.newFixedThreadPool(rapidConfig.getProcessThread(), new ThreadFactoryBuilder().setNameFormat("MpmcProcessor-pool-%d").build());
            for (int i = 0; i < rapidConfig.getProcessThread(); i++) {
                executorService.submit(new ConsumerProcessor());
            }
        } else {
            consumerProcessorThread = new Thread(new ConsumerProcessor());
            consumerProcessorThread.start();
        }
    }

    /**
     * 执行器关闭方法,资源释放
     */
    @Override
    public void shutdown() {
        nettyCoreProcessor.shutdown();
        this.isRunning = false;
        if (usedExecutorPool) {
            executorService.shutdown();
        }
    }

    public class ConsumerProcessor implements Runnable {

        @Override
        public void run() {
            while (isRunning) {
                HttpRequestWrapper event = null;
                try {
                    event = mpmcBlockingQueue.take();
                    nettyCoreProcessor.process(event);
                } catch (Throwable t) {
                    if (event != null) {
                        HttpRequest request = event.getFullHttpRequest();
                        ChannelHandlerContext ctx = event.getCtx();
                        try {
                            log.error("BatchEventProcessorListener onException 请求处理失败, request: {}, errorMessage: {}", request, t.getMessage(), t);
                            FullHttpResponse fullHttpResponse = ResponseHelper.getHttpResponse(ResponseCode.INTERNAL_ERROR);

                            if (!HttpUtil.isKeepAlive(request)) {
                                ctx.writeAndFlush(fullHttpResponse).addListener(ChannelFutureListener.CLOSE);
                            } else {
                                fullHttpResponse.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
                                ctx.writeAndFlush(fullHttpResponse);
                            }
                        } catch (Exception e) {
                            // ignore
                            log.error("ConsumerProcessorListener onException 请求回写失败, request: {}, errorMessage: {}", request, e.getMessage(), e);
                        }
                    } else {
                        log.error("ConsumerProcessorListener onException event is empty, request: {}, errorMessage: {}", t.getMessage(), t);
                    }
                }
            }
        }
    }
}
